package com.pisolution.scheduler.runtime.registry;

import com.pisolution.scheduler.core.utils.OSUtils;
import com.pisolution.scheduler.datasource.command.CommandType;
import com.pisolution.scheduler.datasource.instance.ExecutionState;
import com.pisolution.scheduler.datasource.instance.TaskInstance;
import com.pisolution.scheduler.datasource.instance.WorkflowInstance;
import com.pisolution.scheduler.runtime.master.MasterServer;
import com.pisolution.scheduler.runtime.process.ProcessService;
import com.pisolution.scheduler.runtime.worker.WorkerServer;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Nullable;
import java.util.List;
import java.util.Locale;

/**
 * 调度体系核心运行时容错机制 该容错机制只针对运行时环境中的两类关键节点.
 *
 * <ul>
 *     <li>{@link MasterServer 管理节点}</li>
 *     <li>{@link WorkerServer 工作节点}</li>
 * </ul>
 *
 * @author sora
 * @version 0.1.0
 */
@Component
public class FaultTolerant extends CachedZookeeperClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(FaultTolerant.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于处理管理节点与工作节点的失效转移. */
    private final ProcessService service;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化容错服务.
     *
     * @param config 用于配置注册中心.
     * @param service 用于处理管理节点与工作节点的失效转移.
     */
    public FaultTolerant(final ZookeeperConfig config, final ProcessService service) {
        super(config);

        this.service = service;
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    /**
     * 按以下步骤运行以启用管理节点与工作节点的容错机制.
     *
     * <ol>
     *     <li>竞争调度系统的主控权</li>
     *     <li>在注册中心中初始化系统必备路径 包含管理节点 工作节点以及失效节点 (非必须)</li>
     *     <li>等待管理节点可用</li>
     *     <li>如果是初次启动 则触发一次容错机制以处理前一次遗留的任务</li>
     *     <li>监听注册中心中的管理节点与工作节点的变更并作出响应</li>
     * </ol>
     */
    @SuppressWarnings("BusyWait")
    public void start() {
        // 1. 创建一个分布式互斥锁以锁住用于管理节点容错的路径.
        InterProcessMutex mutex = null;

        try {
            // 2. 初始化该分布式锁.
            mutex = new InterProcessMutex(super.getClient(), "/lock/failover/startup-masters");

            // 3. 竞争锁的控制权.
            mutex.acquire();

            // 4. 初始化系统路径.
            try {
                super.persist("/servers/master", "");
                super.persist("/servers/worker", "");
                super.persist("/dead-servers", "");

                LOGGER.info("Succeed to initialize the system path in the registry center.");
            } catch (Exception e) {
                LOGGER.error("Could not initialize the system path in the registry center.", e);
            }

            // 5. 等待当前节点注册完成.
            while (!this.checkIfNodeExist(OSUtils.getHost(), NodeType.MASTER)) {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException ignore) {
                    // 无需处理该异常.
                }
            }

            // 6. 初次启动管理节点时触发一次失效转移机制 处理前一次运行时遗留的任务.
            if (this.getActiveMasterNum() == 1) {
                this.removeNode(null, NodeType.MASTER);
                this.removeNode(null, NodeType.WORKER);
            }

            // 7. 监听节点变更事件.
            super.registerListener();
        } catch (Exception e) {
            LOGGER.error("Could not enable the fault tolerant strategy for the core runtime of the scheduler system.");
        } finally {
            // 8. 释放锁的控制权.
            this.releaseMutex(mutex);
        }
    }

    /**
     * 检查给定路径是否存在于特定节点路径列表中.
     *
     * @param path 需要检查的节点存储路径.
     * @param type 节点所属的服务类型.
     *
     * @return 如果路径存在则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean checkIfNodeExist(final String path, final NodeType type) {
        // 1. 检查给定路径是否合乎规则.
        if (StringUtils.isBlank(super.getNodePath(type))) {
            LOGGER.error("The path {} to be checked is not exist.", path);

            return false;
        }

        // 2. 检查该类型的服务节点下是否存在该节点.
        return super.getServerMap(type).containsKey(path);
    }

    /**
     * 获取当前处于活动状态的管理节点的数量.
     *
     * @return 处于活动状态的管理节点的数量.
     */
    private int getActiveMasterNum() {
        try {
            // 1. 检查管理节点注册路径是否存在.
            if (super.exists(super.getNodePath(NodeType.MASTER))) {
                return super.children(super.getNodePath(NodeType.MASTER)).size();
            }
        } catch (Exception e) {
            LOGGER.error("Could not iterate the path which stored the master servers.", e);
        }

        return 0;
    }

    @Override
    public void close() {
        super.close();
    }

    // -------------------------------------------------------------------------------------------------
    //  监听节点变更事件
    // -------------------------------------------------------------------------------------------------

    @Override
    protected void changed(CuratorFramework client, CuratorCacheListener.Type type, String path) {
        // 1. 处理管理节点变更事件.
        if (path.startsWith(super.getNodePath(NodeType.MASTER) + "/")) {
            switch (type) {
                case NODE_CREATED:
                    LOGGER.info("New master server {} registered in the registry center.", path);

                    break;
                case NODE_DELETED:
                    this.removeNode(path, NodeType.MASTER);

                    break;
                default:
                    break;
            }
        }

        // 2. 处理工作节点变更事件.
        if (path.startsWith(super.getNodePath(NodeType.WORKER) + "/")) {
            switch (type) {
                case NODE_CREATED:
                    LOGGER.info("New worker server {} registered in the registry center.", path);

                    break;
                case NODE_DELETED:
                    LOGGER.info("The worker server {} removed from the registry center.", path);

                    this.removeNode(path, NodeType.WORKER);
                    break;
                default:
                    break;
            }
        }
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (故障转移)
    // -------------------------------------------------------------------------------------------------

    /**
     * 如果分布式锁存在 则释放分布式锁的控制权.
     *
     * @param mutex 需要被释放控制权的分布式锁.
     */
    public void releaseMutex(@Nullable final InterProcessMutex mutex) {
        // 1. 确认分布式锁是否需要被释放.
        if (mutex == null) {
            return;
        }

        // 2. 尝试释放该锁.
        try {
            mutex.release();
        } catch (Exception e) {
            // 3. 确认该锁释放时机是否得当.
            if (e.getMessage().contains("instance must be started before")) {
                LOGGER.warn("The lock released before acquiring.");

                return;
            }

            LOGGER.error("Could not release the lock.", e);
        }
    }

    /**
     * 从给定类型的服务节点列表中移除指定的节点并触发该节点的失效转移机制.
     *
     * @param path 需要被移除的节点地址.
     * @param type 该节点所属的服务类型.
     */
    private void removeNode(@Nullable String path, NodeType type) {
        LOGGER.info("The node {} of type {} is deleted.", path, type);

        // 1. 创建一个分布式锁以锁住用于实施失效转移的路径.
        InterProcessMutex mutex = null;

        try {
            // 2. 初始化该分布式锁.
            mutex = new InterProcessMutex(this.getClient(), this.getFailoverLockPath(type));

            // 3. 获取该锁的控制权.
            mutex.acquire();

            // 4. 尝试从节点路径中解析出该服务所处的主机信息.
            String host = null;

            // 5. 检验节点路径是否为空.
            if (StringUtils.isNotBlank(path)) {
                host = path.substring(path.lastIndexOf('/') + 1);

                // 6. 校验解析出的地址信息是否为空.
                if (StringUtils.isBlank(host)) {
                    LOGGER.error("Unknown host {}, could not remove the given node.", host);

                    return;
                }

                // 7. 确认该节点在失效节点存储路径下的路径名.
                final String hostPath = String.format("/dead-servers/%s_%s", type.name().toLowerCase(Locale.US), host);

                // 8. 将该节点添加到失效节点路径下.
                if (!super.exists(hostPath)) {
                    super.persist(hostPath, type.name().toLowerCase(Locale.US) + "_" + host);

                    LOGGER.info("Add the server running in the host {} to the dead server path.", host);
                }
            }

            // 9. 触发该节点的失效转移机制.
            switch (type) {
                case MASTER:
                    this.failoverMasterServer(host);

                    break;
                case WORKER:
                    this.failoverWorkerServer(host);

                    break;
                default:
                    break;
            }
        } catch (Exception e) {
            LOGGER.error("Could not remove the node {} of type {} from the registry center.", path, type, e);
        } finally {
            // 10. 释放该锁的控制权.
            this.releaseMutex(mutex);
        }
    }

    /**
     * 获取适用于给定节点类型的失效转移实施路径.
     *
     * @param type 待实施失效转移的节点类型.
     *
     * @return 该节点类型所对应的失效转移实施路径.
     */
    private String getFailoverLockPath(final NodeType type) {
        switch (type) {
            case MASTER:
                return "/lock/failover/masters";
            case WORKER:
                return "/lock/failover/workers";
            default:
                return "";
        }
    }

    /**
     * 遵循以下流程以实施管理节点失效转移.
     *
     * <ol>
     *     <li>查询需要进行恢复的工作流实例 如未指定主机地址 则恢复所有工作流实例</li>
     *     <li>清空原有工作流实例的受管主机地址</li>
     *     <li>追加{@link CommandType#RECOVER_FAULT_TOLERANT 恢复指令}到系统控制指令集中</li>
     * </ol>
     *
     * @param host 需要实施失效转移的管理节点网络地址.
     */
    private void failoverMasterServer(@Nullable final String host) {
        LOGGER.info("Start the failover strategy for the master server {}.", host);

        // 1. 查询所有需要进行失效转移的工作流实例.
        final List<WorkflowInstance> instanceList = this.service.queryWorkflowInstanceNeedFailover(host);

        for (final WorkflowInstance instance: instanceList) {
            // 2. 如果未指定主机信息 则跳过.
            if ("NULL".equals(instance.getHost())) {
                continue;
            }

            this.service.processInstanceNeedFailover(instance);
        }

        LOGGER.info("The failover of the master server {} completed.", host);
    }

    /**
     * 遵循以下流程以实施工作节点失效转移.
     *
     * <ol>
     *     <li>查询需要进行恢复的任务实例 如未指定主机地址 则恢复所有任务实例</li>
     *     <li>逐条确认所需恢复的任务实例是否需要恢复 判定条件为主机地址是否为空与启动时间是否延后</li>
     *     <li>逐条关联相应的工作流实例以便进行环境清理</li>
     *     <li>逐条调用相应的处理器适配器清理前一次运行遗留的环境</li>
     *     <li>逐条更改任务实例状态为{@link ExecutionState#NEED_FAULT_TOLERANCE 需要容错}</li>
     *     <li>保存修改后的任务实例</li>
     * </ol>
     *
     * @param host 需要实施失效转移的工作节点网络地址.
     */
    private void failoverWorkerServer(final String host) {
        LOGGER.info("Start the failover strategy for the worker server {}.", host);

        // 1. 查询所有需要进行失效转移的任务实例.
        final List<TaskInstance> instanceList = this.service.queryTaskInstanceNeedFailover(host);

        for (final TaskInstance task: instanceList) {
            // 2. 校验当前任务实例是否符合实施失效转移的必备条件.
            if (!this.checkTaskInstanceNeedFailover(task)) {
                continue;
            }

            // 3. 填充该任务相关的进程实例信息.
            final WorkflowInstance instance = this.service.findWorkflowInstanceById(task.getWorkflowInstanceId());
            if (instance != null) {
                task.setWorkflowInstance(instance);
            }

            // 4. 清理上一次运行该任务实例时的环境.
            // TODO: 待相应模块添加后完善该逻辑 当前只需要清理yarn job.

            // 5. 更改任务实例的运行状态.
            task.setState(ExecutionState.NEED_FAULT_TOLERANCE);

            // 6. 更新数据库中该任务实例的相关记录.
            this.service.saveTaskInstance(task);
        }

        LOGGER.info("The failover of the worker server {} completed.", host);
    }

    /**
     * 校验给定任务实例是否需要实施失效转移.
     *
     * @param instance 需要校验的任务实例对象.
     *
     * @return 如果需要实施失效转移则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean checkTaskInstanceNeedFailover(final TaskInstance instance) {
        // 1. 如果尚未指定需要运行该任务实例的主机地址 则无需进行失效转移.
        if (StringUtils.isBlank(instance.getHost())) {
            return false;
        }

        // 2. 校验执行该任务实例的主机是否已经注册到注册中心.
        if (this.checkIfNodeExist(instance.getHost(), NodeType.WORKER)) {
            // 3. 如果任务实例在工作节点启动后启动 则无需实施失效转移.
            return !this.checkTaskAfterWorkerStart(instance);
        }

        return true;
    }

    /**
     * 校验任务实例是否在工作节点启动之后才启动.
     *
     * @param instance 需要校验的任务实例.
     *
     * @return 如果在工作节点启动之后才启动则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean checkTaskAfterWorkerStart(final TaskInstance instance) {
        // 1. 获取所有注册到注册中心的工作节点.
        final List<Server> servers = super.getServerList(NodeType.WORKER);

        // 2. 筛选校验给定任务实例所在主机的注册时间(启动时间).
        for (final Server server: servers) {
            if (instance.getHost().equals(server.getHost() + ":" + server.getPort())) {
                return instance.getStartTime().after(server.getCreateTime());
            }
        }

        return false;
    }
}
